package com.market.goods.producer.impl;

import com.market.goods.producer.KafkaClusterProducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Slf4j
@Service
public class KafkaClusterProducerImpl implements KafkaClusterProducer {

    @Autowired
    KafkaTemplate<String, String> kafkaTemplate;

    @Transactional
    @Override
    public void send(String message) {
        kafkaTemplate.send("send-topic", message);
    }

    @Transactional
    @Override
    public void sendToKafkaStandardMessageAsync(String topic,String message) {

        // 构建待发送的消息
        ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, message);

        // 尝试发送消息
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(record);

        // 发送消息回调
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            /**
             * 发送成功回调方法
             */
            @Override
            public void onSuccess(SendResult<String, String> sendResultMap) {
//                log.info("onSuccessSendResult={}", sendResultMap);
            }

            /**
             * 发送失败回调方法
             */
            @Override
            public void onFailure(Throwable throwable) {
                log.info("onFailureThrowable={}", throwable);
            }
        });
    }

    @Transactional
    @Override
    public void sendToKafkaStandardMessageSync(String message) {
        //构建发送消息
        ProducerRecord<String, String> record = new ProducerRecord<String, String>("sync-topic", message);
        try {
            // 使用模板发送消息
            kafkaTemplate.send(record).get(10, TimeUnit.SECONDS);

        } catch (InterruptedException e) {
            log.info("InterruptedException={}", e);

        } catch (ExecutionException e) {
            log.info("ExecutionException={}", e);
        } catch (TimeoutException e) {
            log.info("TimeoutException={}", e);
        }
    }

    int k = 0;

    @Override
    public void executeInTransaction(String message) {
        k++;
        String execute = kafkaTemplate.executeInTransaction(operationsCallback -> {
            try {
                if (k / 2 == 0) {
                    throw new Exception("出错了");
                }
                SendResult<String, String> stringStringSendResult = operationsCallback.send("transaction-topic", message).get();
                log.info("executeInTransaction={}", stringStringSendResult);
                return "发送成功";
            } catch (Exception e) {
                log.info("executeInTransactionError={}", e);
                return "发送失败";
            }
        });
        log.info("executeInTransaction={}", execute);
    }
}
